package data.dblp;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import regular.GraphFileInputFormat;
import regular.Initial;

public class FormatController {

	public static boolean isRunOnServer = false;

	public static void main(String[] args) throws IOException,
			ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();	
		conf.setInt("mapred.task.timeout", 3600000);

		String input = "/user/quyet/partition/input";
		String output = "/user/quyet/partition/output";

		if (isRunOnServer) {
			conf.addResource(new Path(
					"/usr/local/hadoop/conf/core-site.xml"));
			conf.addResource(new Path(
					"/usr/local/hadoop/conf/hdfs-site.xml"));
			input = args[0];
			output = args[1];
			Initial.PATH_ALL_NODE = args[2];

		} else {
			conf.addResource(new Path("/usr/local/hadoop/conf/core-site.xml"));
			conf.addResource(new Path("/usr/local/hadoop/conf/hdfs-site.xml"));
			Initial.PATH_ALL_NODE = "/user/quyet/partition/nodes";
		}		

		Job job = new Job(conf, "FormatController");
		job.setJarByClass(data.dblp.FormatController.class);
		job.setMapperClass(data.dblp.PartitionMapper.class);
		job.setNumReduceTasks(0);

		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		
		job.setInputFormatClass(GraphFileInputFormat.class);
		GraphFileInputFormat.addInputPath(job, new Path(input));
		FileOutputFormat.setOutputPath(job, new Path(output));

		FileSystem hdfs = FileSystem.get(conf);
		if (hdfs.exists(new Path(output))) {
			hdfs.delete(new Path(output), true);
		}

		long startTime = System.currentTimeMillis();
		
		job.waitForCompletion(true);
		
		long stopTime = System.currentTimeMillis();
		long elapsedTime = stopTime - startTime;
		System.out.println("Job's time: " + elapsedTime / 1000 + " s\n");
		
		System.exit(0);
	}

}
